package qdu.edu.com.fushanf4.service.search.country;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import qdu.edu.com.fushanf4.service.count.usertype.search2characters.count.SearchTwoCharactersCountMapper;
import qdu.edu.com.fushanf4.utils.ConfigurationUtil;

import java.io.IOException;
import java.util.List;
import java.util.TreeMap;

/**
 * 该类用来搜索每一行中出现的国家名称并计数
 */
public class SearchCountryMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    private final TreeMap<String,Integer> treeMap = new TreeMap<>();
    private final IntWritable valueOut = new IntWritable();
    private final Text keyOut = new Text();
    private List<String> countryList;
    private static String countryNameInput = null;

    public  static void setSearchCountryPath(String searchCountryPath){
        SearchCountryMapper.countryNameInput = searchCountryPath;
    }

    @Override
    protected void setup(Context context) throws IOException {
        if (null != context.getConfiguration().get(SearchCountryDriver.COUNTRY_NAME_INPUT)){
            setSearchCountryPath(context.getConfiguration().get(SearchCountryDriver.COUNTRY_NAME_INPUT));
        }
        String fileName = "part-r-00000";
        countryList = IOUtils.readLines(FileSystem.get(ConfigurationUtil.getConf()).open(new Path(countryNameInput,fileName)));
        for (String countries : countryList){
            treeMap.put(countries,0);
        }
    }
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        SearchTwoCharactersCountMapper.searchKeyValueInMap(treeMap, countryList, value);
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for (String countryName : countryList){
            if (treeMap.get(countryName) != 0) {
                keyOut.set(countryName);
                valueOut.set(treeMap.get(countryName));
                context.write(keyOut,valueOut);
            }
        }
    }
}
